package net.bwie.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.SneakyThrows;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

/**
 * 需求：每10分钟，实时统计各个区域销售总金额
 * @author xuanyu
 * @date 2025/10/28
 */
public class _01StreamWindowDemo {

	public static void main(String[] args) throws Exception {
		// 1. 执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1) ;
		env.enableCheckpointing(30000L) ;

		// 2. 数据源-source
		MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
			.hostname("node101")
			.port(3306)
			.databaseList("db_flink")
			.tableList("db_flink.foo_orders")
			.username("root")
			.password("123456")
			// todo 将foo_orders表中的数据进行全量采集
			.startupOptions(StartupOptions.initial())
			.deserializer(new JsonDebeziumDeserializationSchema())
			.build();
		DataStreamSource<String> stream = env.fromSource(
			mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source"
		);

		// 3. 数据转换-transformation
		SingleOutputStreamOperator<String> stream31 = stream.assignTimestampsAndWatermarks(
			WatermarkStrategy
				.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
				.withTimestampAssigner(new SerializableTimestampAssigner<String>() {
					private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

					@SneakyThrows
					@Override
					public long extractTimestamp(String element, long recordTimestamp) {
						JSONObject jsonObject = JSON.parseObject(element);
						// 获取after值
						JSONObject afterJsonObject = jsonObject.getJSONObject("after");
						// 获取时间
						String createTimeValue = afterJsonObject.getString("createTime");
						// 转换
						Date date = format.parse(createTimeValue);
						// 时间戳
						return date.getTime();
					}
				})
		);

		SingleOutputStreamOperator<Tuple2<String, Double>> stream32 = stream31.map(
			new MapFunction<String, Tuple2<String, Double>>() {
				@Override
				public Tuple2<String, Double> map(String value) throws Exception {
					JSONObject jsonObject = JSON.parseObject(value);
					// 获取after值
					JSONObject afterJsonObject = jsonObject.getJSONObject("after");
					// 获取areaId
					String areaId = afterJsonObject.getString("areaId");
					// 获取totalMoney
					Double totalMoney = afterJsonObject.getDouble("totalMoney");
					// 返回
					return Tuple2.of(areaId, totalMoney);
				}
			}
		);

		KeyedStream<Tuple2<String, Double>, String> stream33 = stream32.keyBy(tuple -> tuple.f0);

		WindowedStream<Tuple2<String, Double>, String, TimeWindow> stream34 = stream33.window(
			TumblingEventTimeWindows.of(Time.minutes(10))
		);

		SingleOutputStreamOperator<String> stream35 = stream34.apply(
			new WindowFunction<Tuple2<String, Double>, String, String, TimeWindow>() {
				private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
				@Override
				public void apply(String areaId,
				                  TimeWindow window,
				                  Iterable<Tuple2<String, Double>> input,
				                  Collector<String> out) throws Exception {
					// 窗口开始时间和结束时间
					String windowStart = format.format(window.getStart());
					String windowEnd = format.format(window.getEnd());
					// 窗口数据求和
					double sum = 0.0 ;
					for (Tuple2<String, Double> tuple : input) {
						sum += tuple.f1 ;
					}
					// 输出
					String output = windowStart + "," + windowEnd + "," + areaId + "," + sum;
					out.collect(output);
				}
			}
		);

		// 4. 数据接收器-sink
		stream35.print();

		// 5. 触发执行-execute
		env.execute("StreamWindowDemo") ;
	}

}
